package com.baishancloud.log.quality

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.sink.SinkUtil
import com.baishancloud.log.quality.pojo._
import com.baishancloud.log.quality.sink.HttpPostSink
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, WindowedStream}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.slf4j.LoggerFactory

import java.time.Duration
import scala.collection.mutable

class WebTerminal extends Serializable


object WebTerminal extends Serializable {

  private val LOGGER = LoggerFactory.getLogger(classOf[WebTerminal])


  /**
   * 处理web端的数据，包括数据转化、聚合、计算、输出
   *
   * @param parameterTool 参数对象
   * @param stream        数据源
   */
  def webTerminal(parameterTool: ParameterTool, stream: DataStream[String]): Unit = {
    val parse: DataStream[(WebLog, StarRocksWeb)] = stream
      .map(new WebLogParse).name("tuple2").uid("2070399e-c6eb-4309-9172-1003001845b2")

    //明细数据入starRocks
    parse
      .map(_._2).name("StarRocksWeb").uid("38472a97-a259-47b2-a357-550fa90af475")
      .filter(_ != null).name("!=null").uid("8e955eb1-5824-4d6a-9c36-a857e24d6dda")
      .map(JSONUtil.toJsonStr(_)).name("toJsonStr").uid("b3f0685a-1d2d-40b0-83a2-658f3e7c093d")
      .addSink(SinkUtil.starRocksJsonString(parameterTool)).setParallelism(parameterTool.getInt(sinkDorisParallel, 1)).name("starRocks").uid("06fe44e8-28c5-4cc3-9731-e4f63c062f40")

    //比率计算
    val windowStream: WindowedStream[WebLog, Tags, TimeWindow] =
      parse
        .map(_._1).name("WebLog").uid("742385a9-1748-4e6c-bc32-969c73e2bbac")
        .filter(_ != null).name("!=null").uid("cfa540b4-12f9-4494-8095-311187b86153")
        .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMinutes(parameterTool.getLong(maxOutOfOrdernessMinutes, 0))).withTimestampAssigner(new SerializableTimestampAssigner[WebLog] {
          override def extractTimestamp(element: WebLog, recordTimestamp: Long): Long = element.timestamp
        }))
        .keyBy(_.tags)
        .window(TumblingEventTimeWindows.of(Time.minutes(parameterTool.getLong(aggMinute, 5))))

    val timeCardFrameRateResult: DataStream[RateResult] = timeCardFrameRateCalculate(windowStream)
    val vvCardFrameRateResult: DataStream[RateResult] = vvCardFrameRateCalculate(windowStream)
    val loadFailedRateResult: DataStream[RateResult] = loadFailedRateCalculate(windowStream)
    timeCardFrameRateResult
      .union(vvCardFrameRateResult, loadFailedRateResult)
      .addSink(new HttpPostSink(parameterTool.getInt(sinkPostThreshold, Int.MaxValue))).setParallelism(1).name("web live比率").uid("1548290e-b211-4c5a-882d-090c5bc87755")
  }


  /**
   * 时长卡顿率<br>
   * sum(if(category = 5 and code = 1, get_json_object(value, '$.count'), 0)) * 60000 / <br>
   * sum(if(category = 5 and code = 1, get_json_object(value, ‘$.windowSize') , 0))
   */
  private def timeCardFrameRateCalculate(windowStream: WindowedStream[WebLog, Tags, TimeWindow]): DataStream[RateResult] = {
    windowStream.process(new ProcessWindowFunction[WebLog, RateResult, Tags, TimeWindow] {
      override def process(key: Tags, context: Context, elements: Iterable[WebLog], out: Collector[RateResult]): Unit = {
        var countSum: Double = 0
        var windowSizeSum: Double = 0
        elements.foreach(w => {
          try {
            if (w.category == 5 && w.code == 1) {
              countSum += JSONUtil.parseObj(w.value).getDouble("count")
              windowSizeSum += JSONUtil.parseObj(w.value).getDouble("windowSize")
            }
          } catch {
            case _: Exception => LOGGER.error(s"category = 5 而且 code = 1，但是value不是json字符串，value内容为：${w.value}")
          }
        })
        val value: Double = (if (windowSizeSum == 0) 0 else countSum * 60000 / windowSizeSum).formatted("%.6f").toDouble
        out.collect(RateResult(context.window.getStart / 1000, step, metric, value, "", Tags(key, timeCardFrameRate), RateResultFields(countSum * 60000, windowSizeSum)))
      }
    }).name("web live时长卡帧率").uid("0b4a9293-6567-4ba0-a079-37b70b77f0a8")
  }


  /**
   * vv卡顿率<br>
   * 分母为首帧事件数sum(if(category = 1 and code = 4, 1, 0))<br>
   * 分子从category = 5 and code =1 的事件中计算，先按uvid聚合, 看一个uvid是否有卡顿<br>
   * select
   * count(distinct if(buffer_count > 0, uvid, null)) as buffer_vv
   * from
   * (
   * select
   * uvid,
   * sum(if(category = 5 and code =1, get_json_object(value, ‘$.count’), 0)) as buffer_count
   * from table
   * group by uvid
   * ) as t1
   */
  private def vvCardFrameRateCalculate(windowStream: WindowedStream[WebLog, Tags, TimeWindow]): DataStream[RateResult] = {
    windowStream.process(new ProcessWindowFunction[WebLog, RateResult, Tags, TimeWindow] {
      override def process(key: Tags, context: Context, elements: Iterable[WebLog], out: Collector[RateResult]): Unit = {
        var firstFrameCount: Double = 0
        val uvidCount: mutable.Map[String, Int] = mutable.Map[String, Int]()
        elements.foreach(w => {
          try {
            if (w.category == 1 && w.code == 4) firstFrameCount += 1
            if (w.category == 5 && w.code == 1) {
              val count: Int = JSONUtil.parseObj(w.value).getInt("count")
              uvidCount.put(w.uvid, uvidCount.getOrElse(w.uvid, 0) + count)
            }
          } catch {
            case _: Exception => LOGGER.error(s"category = 5 而且 code = 1，但是value不是json字符串，value内容为：${w.value}")
          }
        })
        var bufferVv: Double = 0
        uvidCount.values.foreach(v => {
          if (v > 0) bufferVv += 1
        })
        val value: Double = (if (firstFrameCount == 0) 0 else bufferVv / firstFrameCount).formatted("%.6f").toDouble
        out.collect(RateResult(context.window.getStart / 1000, step, metric, value, "", Tags(key, vvCardFrameRate), RateResultFields(bufferVv, firstFrameCount)))
      }
    }).name("web livevv卡顿率").uid("8fa3ca9f-3bba-40a2-9eb2-577a784e3755")
  }


  /**
   * 加载失败率<br>
   * sum(if(category = 3, 1, 0))/sum(if(category = 2 and code = 2 , 1, 0))
   */
  private def loadFailedRateCalculate(windowStream: WindowedStream[WebLog, Tags, TimeWindow]): DataStream[RateResult] = {
    windowStream.process(new ProcessWindowFunction[WebLog, RateResult, Tags, TimeWindow] {
      override def process(key: Tags, context: Context, elements: Iterable[WebLog], out: Collector[RateResult]): Unit = {
        var count1: Double = 0
        var count2: Double = 0
        elements.foreach(w => {
          if (w.category == 3) count1 += 1
          if (w.category == 2 && w.code == 2) count2 += 1
        })
        val value: Double = (if (count2 == 0) 0 else count1 / count2).formatted("%.6f").toDouble
        out.collect(RateResult(context.window.getStart / 1000, step, metric, value, "", Tags(key, loadFailedRate), RateResultFields(count1, count2)))
      }
    }).name("web live加载失败率").uid("c85d785a-4b27-4494-852e-af2f1a242748")
  }

}


